package com.zhaosc.rx.example;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import afu.org.checkerframework.checker.units.qual.s;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.schedulers.TestScheduler;

public class SchedulerExample {

	@Test
	public void testNewThread() {
		Observable.just("hello", "world").observeOn(Schedulers.newThread()).map((s) -> {
			System.out.println("**map**" + Thread.currentThread().getName());
			return s.toUpperCase();
		}).observeOn(Schedulers.io()).subscribeOn(Schedulers.single()).subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println(Thread.currentThread().getName() + "***" + t);
			}
		});

		SleepUtils.sleep(5000L);
	}

	/**
	 * computation()适用于CPU密集型; io()用于1/0 密集型任务，支持异步阻塞 1/0 操作;
	 * trampoline()表示立即执行，如果当前线程有 任务在执行，则会将其暂停， 等插入进来的新任务执行完成之后，再接着执行原先未完成的任务
	 */
	public void testComputation() {

	}

	public void testObserverOn() {

	}

	@Test
	public void testSubscribeOn() {
		Observable.create(new ObservableOnSubscribe<String>() {

			@Override
			public void subscribe(ObservableEmitter<String> emitter) throws Exception {
				System.out.println("***subscribe***" + Thread.currentThread().getName());
				emitter.onNext("zhaosc");
				emitter.onNext("shuaige");
			}
		}).subscribeOn(Schedulers.newThread()).map((s) -> {
			System.out.println("**map**" + Thread.currentThread().getName());
			return s.toUpperCase();
		}).observeOn(Schedulers.newThread()).subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("***accept***" + Thread.currentThread().getName());
				System.out.println("*****" + t);
			}
		});
	}

	/**
	 * 对于测试一些需要精确时间的任务，
	 *  TestScheduler 还是很有用处的，
	 *  可以节省很多等待的 时间 。
	 */
	@Test
	public void testTestScheduler() {
		TestScheduler scheduler = new TestScheduler();
		scheduler.createWorker().schedule(new Runnable() {

			@Override
			public void run() {
				System.out.println("***hello***");
			}
		});
		scheduler.createWorker().schedule(new Runnable() {

			@Override
			public void run() {
				System.out.println("***400***");
			}
		}, 400, TimeUnit.SECONDS);

		scheduler.createWorker().schedule(new Runnable() {

			@Override
			public void run() {
				System.out.println("***600***");
			}
		}, 600, TimeUnit.SECONDS);
		scheduler.advanceTimeTo(1, TimeUnit.SECONDS);
		System.out.println("**********"+scheduler.now(TimeUnit.MILLISECONDS));
		 scheduler.advanceTimeTo(400, TimeUnit.SECONDS);
		System.out.println("**********"+scheduler.now(TimeUnit.SECONDS));
		 scheduler.advanceTimeTo(600, TimeUnit.SECONDS);
		System.out.println("**********"+scheduler.now(TimeUnit.SECONDS));
	}
}
